{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "8a9a169d",
   "metadata": {},
   "source": [
    "# 单动作多智能体\n",
    "\n",
    "MetaGPT的核心优势也在于轻松灵活地开发一个智能体团队。\n",
    "\n",
    "我们需要三个步骤来建立团队并使其运作：\n",
    "\n",
    "定义每个角色能够执行的预期动作\n",
    "\n",
    "基于标准作业程序（SOP）确保每个角色遵守它。通过使每个角色观察上游的相应输出结果，并为下游发布自己的输出结果，可以实现这一点。\n",
    "\n",
    "初始化所有角色，创建一个带有环境的智能体团队，并使它们之间能够进行交互。\n",
    "\n",
    "内容来自于：\n",
    "[https://docs.deepwisdom.ai/v0.8/zh/guide/tutorials/multi\\_agent\\_101.html](https://docs.deepwisdom.ai/v0.8/zh/guide/tutorials/multi%5C_agent%5C_101.html)\n",
    "\n",
    "## 定义动作\n",
    "\n",
    "我们可以定义三个具有各自动作的Role：\n",
    "\n",
    "SimpleCoder 具有 SimpleWriteCode 动作，接收用户的指令并编写主要代码\n",
    "\n",
    "SimpleTester 具有 SimpleWriteTest 动作，从 SimpleWriteCode 的输出中获取主代码并为其提供测试套件\n",
    "\n",
    "SimpleReviewer 具有 SimpleWriteReview 动作，审查来自 SimpleWriteTest 输出的测试用例，并检查其覆盖范围和质量"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "eb5bf879",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025-02-26 17:16:33.295 | INFO     | metagpt.const:get_metagpt_package_root:29 - Package root set to /tmp/code/wow-agent/notebooks\n"
     ]
    }
   ],
   "source": [
    "import re\n",
    "from metagpt.actions import Action, UserRequirement\n",
    "\n",
    "# 构造写代码的动作\n",
    "\n",
    "def parse_code(rsp):\n",
    "    pattern = r\"```python(.*)```\"\n",
    "    match = re.search(pattern, rsp, re.DOTALL)\n",
    "    code_text = match.group(1) if match else rsp\n",
    "    return code_text\n",
    "\n",
    "\n",
    "class SimpleWriteCode(Action):\n",
    "    PROMPT_TEMPLATE: str = \"\"\"\n",
    "    Write a python function that can {instruction}.\n",
    "    Return ```python your_code_here ```with NO other texts,\n",
    "    your code:\n",
    "    \"\"\"\n",
    "    name: str = \"SimpleWriteCode\"\n",
    "\n",
    "    async def run(self, instruction: str):\n",
    "        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)\n",
    "\n",
    "        rsp = await self._aask(prompt)\n",
    "\n",
    "        code_text = parse_code(rsp)\n",
    "\n",
    "        return code_text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "114acc5e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 构造写测试样例的动作\n",
    "class SimpleWriteTest(Action):\n",
    "    PROMPT_TEMPLATE: str = \"\"\"\n",
    "    Context: {context}\n",
    "    Write {k} unit tests using pytest for the given function, assuming you have imported it.\n",
    "    Return ```python your_code_here ```with NO other texts,\n",
    "    your code:\n",
    "    \"\"\"\n",
    "\n",
    "    name: str = \"SimpleWriteTest\"\n",
    "\n",
    "    async def run(self, context: str, k: int = 3):\n",
    "        prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)\n",
    "\n",
    "        rsp = await self._aask(prompt)\n",
    "\n",
    "        code_text = parse_code(rsp)\n",
    "\n",
    "        return code_text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "0d166c34",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 构造审查代码的动作\n",
    "class SimpleWriteReview(Action):\n",
    "    PROMPT_TEMPLATE: str = \"\"\"\n",
    "    Context: {context}\n",
    "    Review the test cases and provide one critical comments:\n",
    "    \"\"\"\n",
    "\n",
    "    name: str = \"SimpleWriteReview\"\n",
    "\n",
    "    async def run(self, context: str):\n",
    "        prompt = self.PROMPT_TEMPLATE.format(context=context)\n",
    "\n",
    "        rsp = await self._aask(prompt)\n",
    "\n",
    "        return rsp"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2146e80a",
   "metadata": {},
   "source": [
    "## 定义角色\n",
    "\n",
    "在许多多智能体场景中，定义Role可能只需几行代码。对于SimpleCoder，我们做了两件事：\n",
    "\n",
    "1. 使用 set\\_actions 为Role配备适当的 Action，这与设置单智能体相同\n",
    "\n",
    "2. 多智能体操作逻辑：我们使Role \\_watch 来自用户或其他智能体的重要上游消息。回想我们的 SOP，SimpleCoder接收用户指令，这是由 MetaGPT 中的UserRequirement引起的Message。因此，我们添加了 self.\\_watch([UserRequirement])。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "000a4e51",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 构造写代码的角色\n",
    "from metagpt.roles import Role\n",
    "class SimpleCoder(Role):\n",
    "    name: str = \"Alice\"\n",
    "    profile: str = \"SimpleCoder\"\n",
    "\n",
    "    def __init__(self, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "        self._watch([UserRequirement])\n",
    "        self.set_actions([SimpleWriteCode])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "104e0937",
   "metadata": {},
   "source": [
    "与上述相似，对于 SimpleTester，我们：\n",
    "\n",
    "1. 使用 set\\_actions 为SimpleTester配备 SimpleWriteTest 动作\n",
    "\n",
    "2. 使Role \\_watch 来自其他智能体的重要上游消息。回想我们的 SOP，SimpleTester从 SimpleCoder 中获取主代码，这是由 SimpleWriteCode 引起的 Message。因此，我们添加了 self.\\_watch([SimpleWriteCode])。\n",
    "\n",
    "3. 重写 \\_act 函数，就像我们在智能体入门中的单智能体设置中所做的那样。在这里，我们希望SimpleTester将所有记忆用作编写测试用例的上下文，并希望有 5 个测试用例。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "5b5411a6",
   "metadata": {},
   "outputs": [],
   "source": [
    "from metagpt.logs import logger\n",
    "from metagpt.schema import Message\n",
    "\n",
    "class SimpleTester(Role):\n",
    "    name: str = \"Bob\"\n",
    "    profile: str = \"SimpleTester\"\n",
    "\n",
    "    def __init__(self, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "        self.set_actions([SimpleWriteTest])\n",
    "        self._watch([SimpleWriteCode])\n",
    "        # self._watch([SimpleWriteCode, SimpleWriteReview])  # feel free to try this too\n",
    "\n",
    "    async def _act(self) -> Message:\n",
    "        logger.info(f\"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})\")\n",
    "        todo = self.rc.todo\n",
    "\n",
    "        # context = self.get_memories(k=1)[0].content # use the most recent memory as context\n",
    "        context = self.get_memories()  # use all memories as context\n",
    "\n",
    "        code_text = await todo.run(context, k=5)  # specify arguments\n",
    "        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))\n",
    "\n",
    "        return msg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "e21bfa4d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 按照相同的过程定义 SimpleReviewer：\n",
    "class SimpleReviewer(Role):\n",
    "    name: str = \"Charlie\"\n",
    "    profile: str = \"SimpleReviewer\"\n",
    "\n",
    "    def __init__(self, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "        self.set_actions([SimpleWriteReview])\n",
    "        self._watch([SimpleWriteTest])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9448d8bb",
   "metadata": {},
   "source": [
    "创建一个团队并添加角色\n",
    "\n",
    "现在我们已经定义了三个 Role，是时候将它们放在一起了。我们初始化所有角色，设置一个 Team，并hire 它们。\n",
    "\n",
    "运行 Team，我们应该会看到它们之间的协作！\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "8ea58a43",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025-02-26 17:18:02.451 | INFO     | __main__:main:9 - write a function that calculates the product of a list\n",
      "2025-02-26 17:18:02.714 | INFO     | metagpt.team:invest:90 - Investment: $3.0.\n",
      "2025-02-26 17:18:02.717 | INFO     | metagpt.roles.role:_act:391 - Alice(SimpleCoder): to do SimpleWriteCode(SimpleWriteCode)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "```python\n",
      "def product_of_list(lst):\n",
      "    product = 1\n",
      "    for num in lst:\n",
      "        product *= num\n",
      "    return product\n",
      "```"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025-02-26 17:18:03.940 | WARNING  | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.\n",
      "2025-02-26 17:18:03.946 | INFO     | __main__:_act:15 - Bob(SimpleTester): to do SimpleWriteTest(SimpleWriteTest)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "```python\n",
      "def test_product_of_list_empty():\n",
      "    assert product_of_list([]) == 1\n",
      "\n",
      "def test_product_of_list_single_element():\n",
      "    assert product_of_list([5]) == 5\n",
      "\n",
      "def test_product_of_list_positive_numbers():\n",
      "    assert product_of_list([1, 2, 3, 4]) == 24\n",
      "\n",
      "def test_product_of_list_negative_numbers():\n",
      "    assert product_of_list([-1, -2, -3, -4]) == -24\n",
      "\n",
      "def test_product_of_list_mixed_numbers():\n",
      "    assert product_of_list([-1, 2, -3, 4]) == 24\n",
      "```"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025-02-26 17:18:09.171 | WARNING  | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.\n",
      "2025-02-26 17:18:09.179 | INFO     | metagpt.roles.role:_act:391 - Charlie(SimpleReviewer): to do SimpleWriteReview(SimpleWriteReview)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Critical Comment:\n",
      "While the test cases cover a variety of scenarios, including empty lists, single elements, positive numbers, negative numbers, and mixed numbers, there is a potential oversight in the test suite. The function `product_of_list` does not handle the case where the list contains a zero. The product of any list that includes zero should be zero, as multiplying by zero always results in zero. The current test suite does not include a test case to verify this behavior, which could lead to an undetected bug if the function is not implemented to handle zero correctly. Adding a test case for a list containing zero would be a critical addition to ensure the function"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025-02-26 17:18:13.093 | WARNING  | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "'s robustness.\n"
     ]
    }
   ],
   "source": [
    "import asyncio\n",
    "from metagpt.team import Team\n",
    "\n",
    "async def main(\n",
    "    idea: str = \"write a function that calculates the product of a list\",\n",
    "    investment: float = 3.0,\n",
    "    n_round: int = 5,\n",
    "):\n",
    "    logger.info(idea)\n",
    "\n",
    "    team = Team()\n",
    "    team.hire(\n",
    "        [\n",
    "            SimpleCoder(),\n",
    "            SimpleTester(),\n",
    "            SimpleReviewer(),\n",
    "        ]\n",
    "    )\n",
    "\n",
    "    team.invest(investment=investment)\n",
    "    team.run_project(idea)\n",
    "    await team.run(n_round=n_round)\n",
    "await main()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2a18d156",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
